package com.jeffse.light.stream.app.function;

import com.jeffse.light.stream.app.pojo.AvgTemperature;
import com.jeffse.light.stream.app.pojo.Temperature;
import com.jeffse.light.stream.core.data.context.KeyedContext;
import com.jeffse.light.stream.core.data.context.ProcessContext;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ListStateDescriptor;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ValueStateDescriptor;
import com.jeffse.light.stream.core.data.state.ListState;
import com.jeffse.light.stream.core.data.state.ValueState;
import com.jeffse.light.stream.core.function.base.KeyedProcessFunction;
import com.jeffse.light.stream.core.function.base.ProcessFunction;
import com.jeffse.light.stream.core.pojo.OutTag;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class OutPutSideProcessFunction extends ProcessFunction<Temperature, AvgTemperature> {
    OutTag outTag1 = null;
    OutTag outTag2 = null;

    public OutPutSideProcessFunction(OutTag outTag1, OutTag outTag2){
        this.outTag1 = outTag1;
        this.outTag2 = outTag2;
    }

    @Override
    public void processElement(ProcessContext context, Temperature value) throws Exception {
        if(value.getCity().equals("郑州")){
            context.collect(value, outTag2);
        }else{
            context.collect(value, outTag1);
        }
    }
}
